package com.atguigu.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext, StreamingContextState}

/**
  * create by undeRdoG on  2021-06-19  18:02
  * 凡心所向，素履以往，生如逆旅，一苇以航。
  */
object SparkStreaming07_Resume {

  def main(args: Array[String]): Unit = {

    /**
      * 恢复数据
      **/
    val ssc: StreamingContext = StreamingContext.getActiveOrCreate("cp", () => {
      val sparkConf = new SparkConf().setAppName("Streaming").setMaster("local[*]")
      val ssc = new StreamingContext(sparkConf, Seconds(3))


      // 获取端口数据
      val lines: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop102", 9999)

      val wordToOne: DStream[(String, Int)] = lines.map((_, 1))

      /**
        * 窗口的范围必须是采集周期的整数倍
        * 第二个参数为滑动步长
        **/
      val value: DStream[(String, Int)] = wordToOne.window(Seconds(6), Seconds(6))

      value.reduceByKey(_ + _)

      ssc
    })

    ssc.checkpoint("cp")

    // 启动采集器
    ssc.start()



    // 等待采集器关闭
    ssc.awaitTermination() //  该方法会一直将 main 线程 block住


  }

}
